package com.fanli.bigdata.mytest

/**
 * Created by liangdong.wu on 2017/3/2.
 */

import org.apache.log4j.{Level, Logger}
import org.apache.spark.{SparkContext, SparkConf}
import org.apache.spark.ml.evaluation.RegressionEvaluator
import org.apache.spark.ml.recommendation.ALS
import org.apache.spark.sql.{Row, SQLContext}

object FanliAlsTrainigDemo {
  Logger.getLogger("org").setLevel(Level.ERROR)
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("MySpakDemo1").setMaster("local[*]")
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)
    import sqlContext.implicits._
    val ratings = sc.textFile("file:///D:/ratings.dat")
      .map(MovieAlsFun.parseRating)
      .toDF()
    val Array(training, validation, test) = ratings.randomSplit(Array(0.6, 0.2, 0.2))

    // Build the recommendation model using ALS on the training data
    val als = new ALS()
      .setMaxIter(5)
      .setRegParam(0.01)
      .setUserCol("userId")
      .setItemCol("movieId")
      .setRatingCol("rating")
    val model = als.fit(training)

    // Evaluate the model by computing the RMSE on the test data
    val predictions = model.transform(test)

    val evaluator = new RegressionEvaluator()
      .setMetricName("rmse")
      .setLabelCol("rating")
      .setPredictionCol("prediction")
    val rmse = evaluator.evaluate(predictions)
    println(s"Root-mean-square error = $rmse")
    sc.stop()
  }
}
